package mapreduce

import (
	"fmt"
	"sync"
	"time"
)

type MasMr struct {
	globalErr   error
	errChan     chan error
	doneChan    chan bool
	count       int
	timeout     time.Duration
	timeoutChan <-chan time.Time
	Timeout     int
}

type mapReduceFunc func() error

func (mr *MasMr) MapDo(fs ...mapReduceFunc) error {
	//syn使用
	var wg sync.WaitGroup
	mr.errChan = make(chan error)
	mr.doneChan = make(chan bool, len(fs))
	if mr.Timeout == 0 {
		//设置默认值
		mr.Timeout = 50
	}
	mr.timeout = time.Second * time.Duration(mr.Timeout)
	mr.timeoutChan = time.After(mr.timeout) //超时控制，默认10s
	//需要挂载数据，超时控制，错误传递
	//panic处理
	defer func() {
		if r := recover(); r != nil {
			//做处理
			fmt.Println("panic recovery deal")
		}
		//关掉chan
		wg.Wait()
		close(mr.errChan)
		close(mr.doneChan)
	}()

	//开始执行
	for k, fitem := range fs {
		wg.Add(1)
		go func(index int, f mapReduceFunc, subCtx *MasMr) {
			defer func() {
				var r any
				if r = recover(); r != nil {
					fmt.Println("panic recovery deal111")
				}
				//在这里去完成
				//如何判断chan已经关闭
				subCtx.doneChan <- true //这里如果没有缓冲，异常关闭已经没有接收端了，所以会死循环
				wg.Done()
			}()
			err := f()
			if err != nil {
				//换成记录日志
				fmt.Println("error ", err)
				subCtx.errChan <- err
			}
		}(k, fitem, mr)
		//计数
		mr.count++
	}
myselect:
	for {
		select {
		//错误控制
		case mr.globalErr = <-mr.errChan:
			if mr.globalErr != nil {
				fmt.Println("select error break", mr.globalErr)
				break myselect
			}
			//当全部完成
		case <-mr.doneChan:
			mr.count--
			if mr.count <= 0 {
				fmt.Println("select break total finish")
				break myselect
			}
		//超时控制
		case <-mr.timeoutChan:
			{
				fmt.Println("select break timeout")
			}
			break myselect
		}
	}
	//也可以使用waitGroup来进行处理，即使其中有一个协程报错，但也需要通过wg等待所有协程结束，才进行返回
	return mr.globalErr
}
